package org.netwia.networks;

import java.util.NoSuchElementException;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.commons.lang3.builder.ToStringBuilder;
import org.netwia.utils.ValidationUtils;

/**
 * Logical connection between nodes. Allows to set up different behavior on null element.
 * Non null elements are forwarded to the queue if they satisfies the throughput conditions.
 * <p></p>
 * Throughput conditions specifies which non null elements are forwarded to the queue as an input.
 * Implementation internally counts how many non null elements passed since the creation.
 * Whenever new element come this count (1 for the first element) is divided by throughputFactor
 * and if the rest is equal throughputModulo, then element is passed to the queue.
 * Otherwise it is thrown away. Internal counter is then increased by 1.
 *  
 * 
 * @author radek.hecl
 *
 * @param <T> type of the network parameter
 */
public class LogicalQueueConnection<T> implements OutputListener<T>, InputProvider<T> {

    /**
     * Builder object.
     * 
     * @param <T> type of the network parameter
     */
    public static class Builder<T> {

        /**
         * Policy which defines how to behave on processing the null element.
         */
        private NullPolicy nullPolicy;

        /**
         * Factor which divides the baseline for the throughput.
         */
        private Integer throughputFactor;

        /**
         * Modulo which determines whether the element is passed or swallowed.
         */
        private Integer throughputModulo;

        /**
         * Sets policy for null elements.
         * 
         * @param nullPolicy policy for null elements
         * @return this instance
         */
        public Builder<T> setNullPolicy(NullPolicy nullPolicy) {
            this.nullPolicy = nullPolicy;
            return this;
        }

        /**
         * Sets throughput factor. If factor is 2, then every second
         * non null element will be passed.
         * 
         * @param throughputFactor throughput factor
         * @return this instance
         */
        public Builder<T> setThroughputFactor(int throughputFactor) {
            this.throughputFactor = throughputFactor;
            return this;
        }

        /**
         * Sets throughput modulo. If thoughputFactor is 2 and modulo is 0,
         * then only even inputs will go through.
         * 
         * @param throughputModulo throughput modulo
         * @return this instance
         */
        public Builder<T> setThroughputModulo(int throughputModulo) {
            this.throughputModulo = throughputModulo;
            return this;
        }

        /**
         * Builds the result object.
         * 
         * @return created object
         */
        public LogicalQueueConnection<T> build() {
            return new LogicalQueueConnection<T>(this);
        }
    }

    /**
     * Policy which defines how to behave on processing the null element.
     */
    private NullPolicy nullPolicy;

    /**
     * Factor which divides the baseline for the throughput.
     */
    private Integer throughputFactor;

    /**
     * Modulo which determines whether the element is passed or swallowed.
     */
    private Integer throughputModulo;

    /**
     * Queue for elements.
     */
    private Queue<T> queue = new LinkedBlockingQueue<T>();

    /**
     * Counter of non null elements.
     */
    private int counter = 0;

    /**
     * Creates new instance.
     * 
     * @param builder builder object
     */
    public LogicalQueueConnection(Builder<T> builder) {
        nullPolicy = builder.nullPolicy;
        throughputFactor = builder.throughputFactor;
        throughputModulo = builder.throughputModulo;
        guardInvariants();
    }

    /**
     * Guards this object to be consistent. Throws exception if this is not the case.
     */
    private void guardInvariants() {
        ValidationUtils.guardNotNull(nullPolicy, "nullPolicy cannot be null");
        ValidationUtils.guardNotNull(throughputFactor, "throughputFactor cannot be null");
        ValidationUtils.guardNotNull(throughputModulo, "throughputModulo cannot be null");
    }

    @Override
    public boolean isNextInputReady() {
        return !queue.isEmpty();
    }

    @Override
    public T getNextInput() {
        T res = queue.poll();
        if (res == null) {
            throw new NoSuchElementException("no element on the input");
        }
        return res;
    }

    @Override
    public synchronized void outputProduced(T output) {
        if (output == null) {
            switch (nullPolicy) {
            case SWALLOW:
                return;
            case THROW_EXCEPTION:
                throw new IllegalArgumentException("null output is not allowed");
            default:
                throw new RuntimeException("unsupported nullPolicy: " + nullPolicy);
            }
        }
        ++counter;
        if (counter % throughputFactor == throughputModulo) {
            queue.add(output);
        }
    }

    @Override
    public String toString() {
        return ToStringBuilder.reflectionToString(this);
    }

}
